﻿using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Concurrent;
using System.Threading.Tasks;

namespace Fast.Extensions.RabbitMQ
{
    /// <summary>
    /// 消息消费
    /// </summary>
    public class RabbitMqMessageConsumer : IRabbitMqMessageConsumer, ITransientDependency
    {
        /// <summary>
        /// 连接池
        /// </summary>
        protected IConnectionPool ConnectionPool { get; }

        /// <summary>
        /// 信道
        /// </summary>
        protected IModel Channel { get; private set; }

        /// <summary>
        /// 消费回调
        /// </summary>
        protected ConcurrentBag<Func<BasicDeliverEventArgs, Task>> Callbacks { get; }

        /// <summary>
        /// RabbitMQ配置选项
        /// </summary>
        protected FastRabbitMqOptions Options { get; }

        protected ILogger<RabbitMqMessageConsumer> Logger;

        public RabbitMqMessageConsumer(IConnectionPool connectionPool, IOptions<FastRabbitMqOptions> options, ILogger<RabbitMqMessageConsumer> logger)
        {
            ConnectionPool = connectionPool ?? throw new ArgumentNullException(nameof(connectionPool));
            Options = options?.Value ?? throw new ArgumentNullException(nameof(options));
            Logger = logger ?? throw new ArgumentNullException(nameof(logger));

            Callbacks = new ConcurrentBag<Func<BasicDeliverEventArgs, Task>>();
        }

        /// <summary>
        /// 接收消息
        /// </summary>
        /// <param name="callback"></param>
        public void OnMessageReceived(Func<BasicDeliverEventArgs, Task> callback)
        {
            Callbacks.Add(callback);
        }

        /// <summary>
        /// 开始消费
        /// </summary>
        /// <param name="context"></param>
        /// <exception cref="ArgumentNullException"></exception>
        public void StartBasicConsume(BasicConsumeContext context)
        {
            CreateChannel(context.ConnectionName);
            //如果是死信队列
            if (context.MessageType == MessageType.Exception)
            {
                ConfigureDeadLetterQueueName(context);
            }
            else
            {
                DeclareExchangeAndQueue(context);
            }

            Logger.LogTrace("开始 RabbitMQ 消费...");

            if (Channel != null)
            {
                var consumer = new AsyncEventingBasicConsumer(Channel);
                consumer.Received += HandleIncomingMessageAsync;

                Channel.BasicConsume(
                    queue: context.Queue.QueueName,
                    autoAck: false,
                    consumer: consumer
                );
            }
            else
            {
                Logger.LogError("StartBasicConsume无法调用：Channel==null");
            }
        }

        /// <summary>
        /// 异步处理传入消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="eventArgs"></param>
        /// <returns></returns>
        private async Task HandleIncomingMessageAsync(object sender, BasicDeliverEventArgs basicDeliverEventArgs)
        {
            try
            {
                foreach (var callback in Callbacks)
                {
                    await callback(basicDeliverEventArgs);
                }

                Channel.BasicAck(basicDeliverEventArgs.DeliveryTag, multiple: false);
            }
            catch (Exception ex)
            {
                Channel.BasicNack(basicDeliverEventArgs.DeliveryTag, multiple: false, requeue: false);
                Logger.LogWarning(ex, "处理消息时出错");
            }
        }

        /// <summary>
        /// 创建信道
        /// </summary>
        /// <param name="connectionName"></param>
        private void CreateChannel(string connectionName)
        {
            DisposeChannelAsync();

            try
            {
                var connection = ConnectionPool.Get(connectionName);
                if (!connection.IsConnected)
                {
                    connection.TryConnect();
                }

                Channel = connection.CreateModel();
            }
            catch (Exception ex)
            {
                Logger.LogWarning(ex.ToString());
            }
        }

        /// <summary>
        /// 申明交换机和队列
        /// </summary>
        /// <param name="context"></param>
        protected virtual void DeclareExchangeAndQueue(BasicConsumeContext context)
        {
            try
            {
                if (context.EnableDeadLetterExchange)
                {
                    //处理DLX
                    if (!context.Queue.Arguments.TryGetValue("x-dead-letter-exchange", out object dlxExchangeName))
                    {
                        dlxExchangeName = $"{context.Exchange.ExchangeName}.{RabbitMqConsts.ExceptionEventSuffix.ToLower()}";
                        context.Queue.Arguments.Add("x-dead-letter-exchange", dlxExchangeName);
                    }
                    if (!context.Queue.Arguments.TryGetValue("x-dead-letter-routing-key", out object dlxRoutingKey))
                    {
                        dlxRoutingKey = $"{context.RoutingKey}{RabbitMqConsts.ExceptionEventSuffix}";
                        context.Queue.Arguments.Add("x-dead-letter-routing-key", dlxRoutingKey);
                    }
                    var dlxQueueName = $"{context.Queue.QueueName}.{RabbitMqConsts.ExceptionEventSuffix.ToLower()}";

                    Channel.ExchangeDeclare(
                        exchange: dlxExchangeName.ToString(),
                        type: "direct",
                        durable: true,
                        autoDelete: false,
                        arguments: null
                    );
                    Channel.QueueDeclare(
                            queue: dlxQueueName,
                            durable: true,
                            exclusive: false,
                            autoDelete: false,
                            arguments: null
                        );
                    Channel.QueueBind(dlxQueueName, dlxExchangeName.ToString(), dlxRoutingKey.ToString());
                }

                //工作交换机
                Channel.ExchangeDeclare(
                    exchange: context.Exchange.ExchangeName,
                    type: context.Exchange.Type,
                    durable: context.Exchange.Durable,
                    autoDelete: context.Exchange.AutoDelete,
                    arguments: context.Exchange.Arguments
                );
                //工作队列
                Channel.QueueDeclare(
                    queue: context.Queue.QueueName,
                    durable: context.Queue.Durable,
                    exclusive: context.Queue.Exclusive,
                    autoDelete: context.Queue.AutoDelete,
                    arguments: context.Queue.Arguments
                );
                //绑定工作交换机和工作队列
                Channel.QueueBind(
                    queue: context.Queue.QueueName,
                    exchange: context.Exchange.ExchangeName,
                    routingKey: context.RoutingKey);
            }
            catch (Exception ex)
            {
                Logger.LogWarning(ex.ToString());
            }
        }

        /// <summary>
        /// 配置死信队列名
        /// </summary>
        /// <param name="context"></param>
        protected virtual void ConfigureDeadLetterQueueName(BasicConsumeContext context)
        {
            var lastIndexOf = context.Queue.QueueName.LastIndexOf(RabbitMqConsts.ExceptionEventSuffix.ToLower());
            context.Queue.QueueName = $"{context.Queue.QueueName.Substring(0, lastIndexOf)}.{RabbitMqConsts.ExceptionEventSuffix.ToLower()}";
        }
        /// <summary>
        /// 释放Channel
        /// </summary>
        protected virtual void DisposeChannelAsync()
        {
            if (Channel == null)
            {
                return;
            }

            try
            {
                Channel.Dispose();
            }
            catch (Exception ex)
            {
                Logger.LogWarning(ex.ToString());
            }
        }
    }
}